package com.atguigu.gmall.realtime.function;

import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/28 14:51
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor pool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        pool = ThreadPoolUtil.getThreadPool();
        
    }
    
    protected abstract void addDim(Connection phoenixConn,
                                   Jedis redis,
                                   T input,
                                   ResultFuture<T> resultFuture);
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        pool.submit(new Runnable() {
            @Override
            public void run() {
                // 先获取Phoenix连接 redis连接
                Connection phoenixConn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
                Jedis redis = RedisUtil.getRedisClient();
                
                addDim(phoenixConn, redis, input, resultFuture);
                
                try {
                    if (phoenixConn != null) {
                        phoenixConn.close();
                    }
                    if (redis != null) {
                        redis.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
/*
通过线程池, 利用多个客户端去进行网络访问

 */